/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by WangYunlai on 2022/07/01.
//

#pragma once

#include "sql/operator/operator.h"
#include "sql/stmt/select_stmt.h"
#include "rc.h"

class Aggregator
{
public:
  Aggregator(int index, const Field* field, int v=0, bool is_star=false)
    : index_(index), field_(field), static_value_(v), is_star_(is_star) {}

  virtual RC operator()(Tuple* tuple) = 0;
  virtual RC post_process() { return RC::SUCCESS; }

  virtual const Value& value() const = 0;

protected:
  int index_ = 0;
  const Field* field_ = nullptr;
  int static_value_ = 0;
  bool is_star_ = false;
};

class CountAggregator : public Aggregator
{
public:

  CountAggregator(int index, const Field* field, int v=0): Aggregator(index, field, v)
  {
    value_.data = &count_;
    value_.type = INTS;
  }

  RC operator()(Tuple* tuple) override
  {
    ++count_;
    return RC::SUCCESS;
  }

  const Value& value() const
  {
    return value_;
  }

private:
  Value value_;
  int count_ = 0;
};

class AvgAggregator : public Aggregator
{
public:

  AvgAggregator(int index, const Field* field, int v=0): Aggregator(index, field, v)
  {
    f_value_.data = &f_avg_;
    f_value_.type = FLOATS;
    i_value_.data = &i_avg_;
    i_value_.type = INTS;
  }

  RC operator()(Tuple* tuple) override
  {
    TupleCell cell;
    tuple->find_cell(*field_, cell);

    switch (cell.attr_type())
    {
    case INTS:
      i_avg_ += *((int*)(cell.data()));
      break;
    case DATES:
      i_avg_ += *((int*)(cell.data()));
      break;
    case FLOATS:
      f_avg_ += *((float*)(cell.data()));
      break;
    default:
      LOG_ERROR("the field \"%s\" is not aggregatable for \"avg\"", field_->field_name());
      return RC::MISMATCH;
    }

    ++count_;
    return RC::SUCCESS;
  }

  RC post_process()
  {
    if (count_ > 0) {
      i_avg_ /= count_;
      f_avg_ /= count_;
    }
    return RC::SUCCESS;
  }

  const Value& value() const
  {
    return field_->attr_type() == INTS ? i_value_ : f_value_;
  }

private:
  Value f_value_;
  Value i_value_;
  mutable float f_avg_ = 0;
  mutable int i_avg_ = 0;
  int count_ = 0;
};



class MaxAggregator : public Aggregator
{
public:

  MaxAggregator(int index, const Field* field, int v=0): Aggregator(index, field, v)
  {
    i_value_.type = INTS;
    i_value_.data = &i_max_;

    f_value_.type = FLOATS;
    f_value_.data = &f_max_;

    s_value_.type = CHARS;
    s_value_.data = (void*)s_max_.c_str();
  }

  RC operator()(Tuple* tuple) override
  {
    TupleCell cell;
    tuple->find_cell(*field_, cell);

    switch (cell.attr_type())
    {
    case INTS:
      i_max_ = std::max(*((int*)(cell.data())), i_max_);
      break;
    case FLOATS:
      f_max_ = std::max(*((float*)(cell.data())), f_max_);
      break;
    case DATES:
      i_max_ = std::max(*((int*)(cell.data())), i_max_);
      break;
    case CHARS:
      {
        const char* cell_data = cell.data();
        if (strcmp(cell_data, s_max_.c_str()) > 0) {
          s_max_ = cell_data;
        }
      }
      break;
    default:
      return RC::MISMATCH;
    }

    return RC::SUCCESS;
  }

  RC post_process()
  {
    s_value_.data = (void*)s_max_.c_str();
  }

  const Value& value() const
  {
    switch(field_->attr_type()) {
      case INTS: return i_value_;
      case FLOATS: return f_value_;
      case DATES: return i_value_;
      default: return s_value_;
    }
  }

private:
  Value f_value_;
  Value i_value_;
  Value s_value_;
  float f_max_ = std::numeric_limits<float>::min();
  int i_max_ = std::numeric_limits<int>::min();
  std::string s_max_;
};


class MinAggregator : public Aggregator
{
public:

  MinAggregator(int index, const Field* field, int v=0): Aggregator(index, field, v)
  {
    i_value_.type = INTS;
    i_value_.data = &i_min_;

    f_value_.type = FLOATS;
    f_value_.data = &f_min_;

    s_value_.type = CHARS;
    s_value_.data = (void*)s_min_.c_str();
  }

  RC operator()(Tuple* tuple) override
  {
    TupleCell cell;
    tuple->find_cell(*field_, cell);

    switch (cell.attr_type())
    {
    case INTS:
      i_min_ = std::min(*((int*)(cell.data())), i_min_);
      break;
    case FLOATS:
      f_min_ = std::min(*((float*)(cell.data())), f_min_);
      break;
    case DATES:
      i_min_ = std::min(*((int*)(cell.data())), i_min_);
      break;
    case CHARS:
      {
        const char* cell_data = cell.data();
        if (!s_min_init) {
          s_min_init = true;
          s_min_ = cell_data;
        }
        else if (strcmp(cell_data, s_min_.c_str()) < 0) {
          s_min_ = cell_data;
        }
      }
      break;
    default:
      return RC::MISMATCH;
    }

    return RC::SUCCESS;
  }

  RC post_process()
  {
    s_value_.data = (void*)s_min_.c_str();
  }

  const Value& value() const
  {
    switch(field_->attr_type()) {
      case INTS: return i_value_;
      case FLOATS: return f_value_;
      default: return s_value_;
    }
  }

private:
  Value f_value_;
  Value i_value_;
  Value s_value_;
  float f_min_ = std::numeric_limits<float>::max();
  int i_min_ = std::numeric_limits<int>::max();
  bool s_min_init = false;
  std::string s_min_;
};


class AggregateOperator : public Operator
{
public:
  AggregateOperator(SelectStmt *select_stmt) : select_stmt_(select_stmt)
  {}

  virtual ~AggregateOperator();

  RC add_aggregator(AggrFunc func, const Field* field, float value=0);

  RC open() override;
  RC next() override;
  RC close() override;

  int tuple_cell_num() const
  {
    return tuple_.cell_num();
  }

  RC tuple_cell_spec_at(int index, const TupleCellSpec *&spec) const;

  Tuple * current_tuple() override;

private:
  SelectStmt *select_stmt_ = nullptr;
  ProjectTuple tuple_;
  std::vector<Aggregator*> aggregators_;
};
